package com.cloud.message.service.impl;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.cloud.message.cache.ConsumerCache;
import com.cloud.message.cache.MessageDelayCache;
import com.cloud.message.dao.MessageDelayDao;
import com.cloud.message.entity.MessageDelay;
import com.cloud.message.rocketmq.MQMessage;
import com.cloud.message.rocketmq.RocketMQSender;
import com.cloud.message.service.MessageDelayService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

@Service
public class MessageDelayServiceImpl extends ServiceImpl<MessageDelayDao, MessageDelay> implements MessageDelayService {

    @Autowired
    private MessageDelayCache messageDelayCache;

    @Autowired
    private ConsumerCache consumerCache;

    @Autowired
    private RocketMQSender rocketMQSender;

    private ThreadPoolExecutor executor;

    @Override
    public boolean save(MessageDelay messageDelay) {
        super.save(messageDelay);
        messageDelayCache.save(messageDelay);
        if (executor == null) {
            synchronized (this) {
                int poolSize = Runtime.getRuntime().availableProcessors();
                executor = new ThreadPoolExecutor(poolSize, poolSize * 4, 180, TimeUnit.SECONDS, new LinkedBlockingQueue<>(poolSize), new ThreadPoolExecutor.CallerRunsPolicy());
            }
        }
        return true;
    }

    @Override
    public void handle() throws Exception {
        if (executor == null) {
            return;
        }
        while (true) {
            Set<String> topics = consumerCache.getTopics();
            if (CollectionUtils.isEmpty(topics) || execute(topics)) {
                return;
            }
        }
    }

    private boolean execute(Set<String> topics) throws Exception {
        AtomicBoolean isContinue = new AtomicBoolean(false);
        CountDownLatch countDownLatch = new CountDownLatch(topics.size());
        topics.forEach(topic -> {
            executor.execute(() -> {
                List<Long> messageIds = messageDelayCache.getNeedHandles(topic, 100);
                if (!CollectionUtils.isEmpty(messageIds)) {
                    List<Long> removedIdList = new ArrayList<>(messageIds.size());
                    Collection<MessageDelay> list = listByIds(messageIds);
                    for (MessageDelay messageDelay : list) {
                        MQMessage message = new MQMessage();
                        message.setLevel(messageDelay.getLevel());
                        message.setTopic(messageDelay.getTopic());
                        message.setContent(messageDelay.getContent());
                        message.setHashKey(messageDelay.getHashKey());
                        message.setUserVersion(messageDelay.getUserVersion());
                        message.setUserSession(messageDelay.getUserSession());
                        if (rocketMQSender.sendMessage(message)) {
                            removedIdList.add(messageDelay.getId());
                        }
                    }
                    messageDelayCache.remove(topic, removedIdList);
                    removeByIds(removedIdList);
                    if (removedIdList.size() > 0) {
                        isContinue.set(true);
                    }
                }
                countDownLatch.countDown();
            });
        });
        countDownLatch.await();
        return isContinue.get();
    }
}
